package com.ibm.util;

import java.io.File;
import java.net.URL;
import java.net.URLDecoder;
import java.util.Properties;

import org.apache.commons.configuration.ConfigurationConverter;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;
import org.apache.storm.guava.base.Throwables;

import com.google.inject.Binder;
import com.google.inject.CreationException;
import com.google.inject.Guice;
import com.google.inject.Inject;
import com.google.inject.Module;
import com.google.inject.name.Named;
import com.google.inject.name.Names;
import com.ibm.util.exception.CPSException;

/**
 * @author Yaoliang Chen
 * @version
 * 
 */
public class LocalConfig {

	private static Logger logger = Logger.getLogger(LocalConfig.class);

	@Inject(optional = true)
	@Named("FEEDBACK_WAITING_SECONDS")
	public static int FEEDBACK_WAITING_SECONDS = 10;

	@Inject(optional = true)
	@Named("REDIS_MAX_CONNECTIONS_NUM")
	public static int REDIS_MAX_CONNECTIONS_NUM = 1024;

	@Inject(optional = true)
	@Named("REDIS_WAITING_SECONDS")
	public static int REDIS_WAITING_SECONDS = 10;

	@Inject(optional = true)
	@Named("WORKER_NUM")
	public static int WORKER_NUM = 1;

	@Inject(optional = true)
	@Named("REDIS_SERVER_HOST")
	public static String REDIS_SERVER_HOST = "9.186.88.253";

	@Inject(optional = true)
	@Named("REDIS_SERVER_PORT")
	public static int REDIS_SERVER_PORT = 6379;

	@Inject(optional = true)
	@Named("PERSISTENCY_DATABASE")
	public static String PERSISTENCY_DATABASE = "test";

	@Inject(optional = false)
	@Named("ZOOKEEPER_HOST_PORT")
	public static String ZOOKEEPER_HOST_PORT = "9.186.88.253:2181";

	@Inject(optional = false)
	@Named("KAFKA_BROKERS")
	public static String KAFKA_BROKERS = "9.186.88.253:9092";

	@Inject(optional = true)
	@Named("STORM_PROCESSOR_SPOUT_NUM")
	public static int STORM_PROCESSOR_SPOUT_NUM = 1;

	@Inject(optional = true)
	@Named("STORM_DATASOURCE_SPOUT_NUM")
	public static int STORM_DATASOURCE_SPOUT_NUM = 1;

	@Inject(optional = true)
	@Named("STORM_DATASOURCE_BOLT_NUM")
	public static int STORM_DATASOURCE_BOLT_NUM = 1;

	@Inject(optional = true)
	@Named("STORM_MESSAGE_SPOUT_NUM")
	public static int STORM_MESSAGE_SPOUT_NUM = 2;

	@Inject(optional = true)
	@Named("STORM_DISPATCH_BOLT_NUM")
	public static int STORM_DISPATCH_BOLT_NUM = 1;

	@Inject(optional = true)
	@Named("STORM_PROCESSOR_BOLT_NUM")
	public static int STORM_PROCESSOR_BOLT_NUM = 1;

	@Inject(optional = true)
	@Named("PROCESSOR_BOLT_TASK_NUM")
	public static int PROCESSOR_BOLT_TASK_NUM = 1;

	@Inject(optional = true)
	@Named("STORM_SEND_BOLT_NUM")
	public static int STORM_SEND_BOLT_NUM = 1;

	@Inject(optional = true)
	@Named("CPS_MODE")
	public static String CPS_MODE = "local";

	@Inject(optional = true)
	@Named("BACKEND_IMPLEMENTATION")
	public static String BACKEND_IMPLEMENTATION = "storm";

	@Inject(optional = true)
	@Named("MESSAGE_QUEUE_USER")
	public static String MESSAGE_QUEUE_USER = "rollen";

	@Inject(optional = true)
	@Named("MESSAGE_QUEUE_PASSWORD")
	public static String MESSAGE_QUEUE_PASSWORD = "root";

	@Inject(optional = true)
	@Named("FROMEMAIL")
	public static String FROMEMAIL = "realtime_ibm@sina.com";

	@Inject(optional = true)
	@Named("FROMEMAILPASSWORD")
	public static String FROMEMAILPASSWORD = "realtime_ibm";

	@Inject(optional = true)
	@Named("MAIL_SMTP_HOST")
	public static String MAIL_SMTP_HOST = "smtp.sina.com";

	@Inject(optional = true)
	@Named("MAIL_SMTP_PORT")
	public static int MAIL_SMTP_PORT = 25;

	@Inject(optional = true)
	@Named("MONGO_CONNECT_STRING")
	public static String MONGO_CONNECT_STRING = "mongodb://9.186.107.15:27017,9.186.107.28:27017,9.186.107.31/?replicaSet=testdb";

	@Inject(optional = true)
	@Named("MASTER_NODE_HOST")
	public static String MASTER_NODE_HOST = "9.186.107.15";

	@Inject(optional = true)
	@Named("MASTER_NODE_PORT")
	public static int MASTER_NODE_PORT = 9091;

	@Inject(optional = true)
	@Named("GRAPHITE_SOCKET_TIMEOUT")
	public static String GRAPHITE_SOCKET_TIMEOUT = "10000";

	@Inject(optional = true)
	@Named("GRAPHITE_CONNECTION_REQUEST_TIMEOUT")
	public static String GRAPHITE_CONNECTION_REQUEST_TIMEOUT = "10000";

	@Inject(optional = true)
	@Named("GRAPHITE_CONNECT_TIMEOUT")
	public static String GRAPHITE_CONNECT_TIMEOUT = "10000";
	@Inject(optional = true)
	@Named("GRAPHITE_INTERVAL")
	public static String GRAPHITE_INTERVAL = "10s";

	@Inject(optional = true)
	@Named("KAFKA_MAX_RETRIES")
	public static String KAFKA_MAX_RETRIES = "3";

	@Inject(optional = true)
	@Named("ConsoleLogLevelThreshold")
	public static String ConsoleLogLevelThreshold = "INFO";
	
	@Inject(optional = true)
	@Named("TOPOLOGY_START_UP_SHELL_PATH")
	public static String TOPOLOGY_START_UP_SHELL_PATH = "./";

	@Inject(optional = true)
	@Named("SPARK_JAR_PATH")
	public static String SPARK_JAR_PATH = "./";
	
	@Inject(optional = true)
	@Named("FORWARDER_START_UP_SHELL_PATH")
	public static String FORWARDER_START_UP_SHELL_PATH = "./";
	
	@Inject(optional = true)
	@Named("SPARK_DEPENDENCY_PATH")
	public static String SPARK_DEPENDENCY_PATH = SPARK_JAR_PATH;
	
	@Inject(optional = true)
	@Named("YARN_RESOURCE_MANAGER_URL")
	public static String YARN_RESOURCE_MANAGER_URL = "http://9.186.107.15:9088/ws/v1/cluster/apps";

	@Inject(optional = true)
	@Named("STORM_CLUSTER_UI_URL")
	public static String STORM_CLUSTER_UI_URL = "http://9.186.107.15:8181/";
	
	
	@Inject(optional = true)
	@Named("FORWARDER_CHECK_SHELL_PATH")
	public static String FORWARDER_CHECK_SHELL_PATH = "./";
	
	@Inject(optional = true)
	@Named("TASK_PER_SPARK_TOPOLOGY")
	public static int TASK_PER_SPARK_TOPOLOGY = 1;
	
	@Inject(optional = true)
	@Named("DEFAULT_DATASOURCE")
	public static String DEFAULT_DATASOURCE = "seyren_sql";
	
	@Inject(optional = true)
	@Named("DEFAULT_PUBLISHER")
	public static String DEFAULT_PUBLISHER = "seyren_sql";

	@Inject(optional = true)
	@Named("DEFAULT_PUBLISHER")
	public static int DEFAULT_SQL_MODEL = 0;

	@Inject(optional = true)
	@Named("KAFKA_PARTITION")
	public static int KAFKA_PARTITION=10;

	@Inject(optional = true)
	@Named("KAFKA_REPLICATIONFACTOR")
	public static int KAFKA_REPLICATIONFACTOR = 2;
	static public boolean init(final Properties prop) {
		try {
			Guice.createInjector(new Module() {
				@Override
				public void configure(Binder binder) {
					Names.bindProperties(binder, prop);
					binder.requestStaticInjection(LocalConfig.class);
				}
			});
			logger = Logger.getLogger(LocalConfig.class);
		} catch (CreationException e) {
			Throwables.propagate(e);
			return false;
		}
		return true;
	}

	public static Properties getLoggerProperties() throws CPSException {
		Properties properties = new Properties();
		try {
			properties = ConfigurationConverter
					.getProperties(new PropertiesConfiguration(LocalConfig.class.getResource("log.properties")));
			properties.put("log4j.appender.appender1.Threshold", ConsoleLogLevelThreshold);

		} catch (ConfigurationException e) {
			System.out.println("Cannot find log.properties");
			logger.error(Throwables.getStackTraceAsString(e));
			Throwables.propagate(e);
		}
		return properties;
	}

	static public void init() {
		Properties properties = null;
		String fileName = getProjectPath() + "/localconfig.conf";
		System.out.println("Reading config from " + fileName);
		logger.info("Reading config from " + fileName);
		if (new File(fileName).exists()) {
			try {
				properties = ConfigurationConverter
						.getProperties(new PropertiesConfiguration(getProjectPath() + "/localconfig.conf"));
			} catch (ConfigurationException e) {
				logger.error(e.getMessage());
				Throwables.propagate(e);
			}

		} else {
			logger.info("Could not find the config file, use the default settings ...");
			System.out.println("Could not find the config file, use the default settings in "
					+ LocalConfig.class.getResource("localconfig.conf"));
			try {
				properties = ConfigurationConverter
						.getProperties(new PropertiesConfiguration(LocalConfig.class.getResource("localconfig.conf")));

			} catch (ConfigurationException e) {
				logger.error(e.getMessage());
				Throwables.propagate(e);
			}
		}
		LocalConfig.init(properties);
	}

	static {
		logger.info("Running init");
		System.out.println("Running init");
		init();
	}

	public static String getProjectPath() {
		try {
			URL url = LocalConfig.class.getProtectionDomain().getCodeSource().getLocation();

			String filePath = URLDecoder.decode(url.getPath(), "utf-8");

			// the directory of jar
			if (filePath.endsWith(".jar")) {
				filePath = filePath.substring(0, filePath.lastIndexOf("/") + 1);
			}
			File file = new File(filePath);
			filePath = file.getAbsolutePath();
			return filePath;
		} catch (Exception e) {

		}
		return null;
	}

	public static void main(String[] args) throws CPSException {

		Properties properties = LocalConfig.getLoggerProperties();
		PropertyConfigurator.configure(properties);
		System.out.println(
				"log4j.appender.appender1.Threshold: " + properties.getProperty("log4j.appender.appender1.Threshold"));

		logger = Logger.getLogger(LocalConfig.class);
		// Appender appender = logger.getAppender("appender1");
		// System.out.println(appender.toString());
		logger.debug("This is debug.");
		logger.info("This is info.");
	}

}
